package com.rocketmq.aliyun.produce;

import com.alibaba.fastjson.JSONObject;
import org.apache.rocketmq.client.AccessChannel;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.common.RemotingHelper;

import java.util.*;

/**
 * 顺序消息
 *
 * @author zhangyan
 * @version $ Id: SequenceProducerMessage.java, v 0.1 2021/5/24 20:14 zhangyan Exp $
 */
public class SequenceProducerMessage {
    public static void main(String[] args) throws Exception {
        /**
         * 创建Producer，并开启消息轨迹。
         * 如果不想开启消息轨迹，可以按照以下方式创建：
         * DefaultMQProducer producer = new DefaultMQProducer("YOUR ORDER GROUP ID", getAclRPCHook());
         */
        DefaultMQProducer producer = new DefaultMQProducer("groupSequence", true, null);

        /**
         * 设置使用接入方式为阿里云，在使用云上消息轨迹的时候，需要设置此项，如果不开启消息轨迹功能，则运行不设置此项。
         */
        producer.setAccessChannel(AccessChannel.LOCAL);
        /**
         * 设置为您从阿里云控制台获取的接入点信息，类似“http://MQ_INST_XXXX.aliyuncs.com:80”。
         */
        producer.setNamesrvAddr("127.0.0.1:9876");
        producer.start();
        List<Map> arr = new ArrayList<>();
        for (int i = 111; i < 113; i++) {
            Map<Integer, String> map1 = new HashMap<>();
            map1.put(i * 100, i * 100 + "创建订单");
            Map<Integer, String> map2 = new HashMap<>();
            map2.put(i * 100, i * 100 + "支付订单");
            Map<Integer, String> map3 = new HashMap<>();
            map3.put(i * 100, i * 100 + "发货订单");
            arr.add(map1);
            arr.add(map2);
            arr.add(map3);
        }

        for (Map map : arr) {
            Set set = map.keySet();
            Integer orderId = 0;
            for (Object o : set) {
                orderId = (Integer) o;
            }
            try {
                Message msg = new Message("topicSequence",
                        "tagSequence",
                        "aaaa",
                        JSONObject.toJSONString(map).getBytes(RemotingHelper.DEFAULT_CHARSET));

                SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
                    @Override
                    public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                        // 选择适合自己的分区选择算法，保证同一个参数得到的结果相同。
                        Integer id = (Integer) arg;
                        int index = id % mqs.size();
                        return mqs.get(index);
                    }
                }, orderId);
                System.out.println("发送结果：" + sendResult);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        producer.shutdown();
    }
}
